package com.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.gmall.realtime.common.base.BaseApp;
import com.gmall.realtime.common.bean.UserRegisterBean;
import com.gmall.realtime.common.constant.Constant;
import com.gmall.realtime.common.function.DorisMapFunction;
import com.gmall.realtime.common.util.DateFormatUtil;
import com.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class DwsUserUserRegisterWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsUserUserRegisterWindow().start(
                10025,
                1,
                "dws_user_user_register_window",
                Constant.TOPIC_DWD_USER_REGISTER
        );
    }

    /**
     * 处理用户注册窗口统计的主方法
     *
     * @param env Flink流处理执行环境
     * @param dataStreamSource 从Kafka获取的原始数据流
     */
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {
        // 1. 数据ETL处理：将原始JSON字符串转换为UserRegisterBean对象
        SingleOutputStreamOperator<UserRegisterBean> etlStream = etl(dataStreamSource);

        // 2. 添加水位线并进行窗口聚合计算
        // 使用10秒的滚动事件时间窗口，统计注册用户数
        SingleOutputStreamOperator<UserRegisterBean> reduceStream = getReduceStream(etlStream);

        // 调试用：打印聚合结果到控制台
        // reduceStream.print();

        // 3. 将结果写入Doris数据库
        // 先通过DorisMapFunction将Bean对象转换为Doris可识别的格式
        // 然后使用FlinkSinkUtil获取Doris sink并写入
        reduceStream
                .map(new DorisMapFunction<>())
                .sinkTo(
                        FlinkSinkUtil.getDorisSink(
                                Constant.DORIS_DATABASE + "." + Constant.DWS_USER_USER_REGISTER_WINDOW,
                                Constant.DWS_USER_USER_REGISTER_WINDOW
                        )
                );
    }


    /**
     * 获取用户注册窗口统计的聚合流
     * 1. 为数据流分配水位线和事件时间
     * 2. 使用10秒的滚动窗口进行聚合计算
     * 3. 在窗口函数中设置窗口起止时间和当前日期
     *
     * @param etlStream 经过ETL处理的用户注册数据流
     * @return 聚合后的用户注册统计结果流
     */
    private SingleOutputStreamOperator<UserRegisterBean> getReduceStream(SingleOutputStreamOperator<UserRegisterBean> etlStream) {
        // 分配水位线和事件时间
        SingleOutputStreamOperator<UserRegisterBean> reduceStream = etlStream
                // 设置最大乱序时间为3秒
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<UserRegisterBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                // 从UserRegisterBean中提取时间戳作为事件时间
                                .withTimestampAssigner(new SerializableTimestampAssigner<UserRegisterBean>() {
                                    @Override
                                    public long extractTimestamp(UserRegisterBean userRegisterBean, long l) {
                                        return DateFormatUtil.dateTimeToTs(userRegisterBean.getCurDate());
                                    }
                                }))
                // 使用10秒的滚动事件时间窗口
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(10)))
                // 窗口聚合计算
                .reduce(
                        // Reduce函数：累加注册用户数
                        new ReduceFunction<UserRegisterBean>() {
                            @Override
                            public UserRegisterBean reduce(UserRegisterBean value1, UserRegisterBean value2) throws Exception {
                                value1.setRegisterCt(value1.getRegisterCt() + value2.getRegisterCt());
                                return value1;
                            }
                        },
                        // 窗口函数：设置窗口元数据信息
                        new AllWindowFunction<UserRegisterBean, UserRegisterBean, TimeWindow>() {
                            @Override
                            public void apply(TimeWindow timeWindow, Iterable<UserRegisterBean> iterable, Collector<UserRegisterBean> collector) throws Exception {
                                // 获取窗口起止时间
                                long start = timeWindow.getStart();
                                long end = timeWindow.getEnd();

                                // 获取聚合结果
                                UserRegisterBean userRegisterBean = iterable.iterator().next();

                                // 设置窗口统计信息
                                userRegisterBean.setStt(DateFormatUtil.tsToDateTime(start));  // 窗口开始时间
                                userRegisterBean.setEdt(DateFormatUtil.tsToDateTime(end));    // 窗口结束时间
                                userRegisterBean.setCurDate(DateFormatUtil.tsToDate(System.currentTimeMillis()));  // 当前处理时间

                                collector.collect(userRegisterBean);
                            }
                        });
        return reduceStream;
    }


    /**
     * 对Kafka原始数据进行ETL处理，转换为UserRegisterBean对象流
     * 1. 解析JSON字符串
     * 2. 过滤空数据
     * 3. 提取创建时间字段
     * 4. 构建用户注册统计对象
     *
     * @param kafkaSource 从Kafka获取的原始数据流，包含用户注册信息的JSON字符串
     * @return 转换后的UserRegisterBean数据流，每个注册用户对应一个计数为1的记录
     */
    private SingleOutputStreamOperator<UserRegisterBean> etl(DataStreamSource<String> kafkaSource) {
        return kafkaSource.flatMap(new FlatMapFunction<String, UserRegisterBean>() {
            @Override
            public void flatMap(String s, Collector<UserRegisterBean> collector) throws Exception {
                try {
                    // 将JSON字符串解析为JSONObject
                    JSONObject jsonObject = JSON.parseObject(s);
                    // 检查JSON对象是否非空
                    if (jsonObject.size() > 0) {
                        // 从JSON中提取用户注册时间字段
                        String createTime = jsonObject.getString("create_time");
                        // 构建用户注册统计对象，每个用户注册记为1次
                        collector.collect(UserRegisterBean.builder().registerCt(1L).curDate(createTime).build());
                    }
                } catch (Exception e) {
                    // 异常处理：打印异常堆栈，但不中断流程
                    e.printStackTrace();
                }

            }
        });
    }


}